package com.seedog.mq.core;

import com.seedog.mq.model.Message;
import com.seedog.mq.exception.TopicNotFoundException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 消息队列实现类，提供主题管理和消息路由功能
 */
public class MessageQueueImpl implements MessageQueue {
    private static final Logger logger = LoggerFactory.getLogger(MessageQueueImpl.class);
    
    private final Map<String, Topic> topics;
    private final ReadWriteLock topicsLock;
    private boolean running = true;
    
    public MessageQueueImpl() {
        this.topics = new ConcurrentHashMap<>();
        this.topicsLock = new ReentrantReadWriteLock();
        logger.info("MessageQueue initialized");
    }
    
    @Override
    public boolean createTopic(String topicName) {
        checkRunning();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        topicsLock.writeLock().lock();
        try {
            if (topics.containsKey(topicName)) {
                logger.warn("Topic '{}' already exists", topicName);
                return false;
            }
            
            Topic topic = new TopicImpl(topicName);
            topics.put(topicName, topic);
            logger.info("Topic '{}' created successfully", topicName);
            return true;
        } finally {
            topicsLock.writeLock().unlock();
        }
    }
    
    @Override
    public boolean deleteTopic(String topicName) {
        checkRunning();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        topicsLock.writeLock().lock();
        try {
            Topic topic = topics.remove(topicName);
            if (topic == null) {
                logger.warn("Topic '{}' not found for deletion", topicName);
                return false;
            }
            
            topic.close();
            logger.info("Topic '{}' deleted successfully", topicName);
            return true;
        } finally {
            topicsLock.writeLock().unlock();
        }
    }
    
    @Override
    public boolean topicExists(String topicName) {
        checkRunning();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        topicsLock.readLock().lock();
        try {
            return topics.containsKey(topicName);
        } finally {
            topicsLock.readLock().unlock();
        }
    }
    
    @Override
    public Set<String> getAllTopics() {
        checkRunning();
        
        topicsLock.readLock().lock();
        try {
            return new HashSet<>(topics.keySet());
        } finally {
            topicsLock.readLock().unlock();
        }
    }
    
    @Override
    public String sendMessage(String topicName, Message message) throws TopicNotFoundException {
        checkRunning();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        if (message == null) {
            throw new IllegalArgumentException("Message cannot be null");
        }
        
        Topic topic = getTopic(topicName);
        return topic.sendMessage(message);
    }
    
    @Override
    public List<String> sendMessages(String topicName, List<Message> messages) throws TopicNotFoundException {
        checkRunning();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        if (messages == null || messages.isEmpty()) {
            throw new IllegalArgumentException("Messages cannot be null or empty");
        }
        
        Topic topic = getTopic(topicName);
        return topic.sendMessages(messages);
    }
    
    @Override
    public Message pollMessage(String topicName, String consumerId, long timeout) throws TopicNotFoundException {
        checkRunning();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        if (consumerId == null || consumerId.isEmpty()) {
            throw new IllegalArgumentException("Consumer ID cannot be null or empty");
        }
        
        Topic topic = getTopic(topicName);
        return topic.pollMessage(consumerId, timeout);
    }
    
    @Override
    public boolean acknowledge(String topicName, String consumerId, String messageId) throws TopicNotFoundException {
        checkRunning();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        if (consumerId == null || consumerId.isEmpty()) {
            throw new IllegalArgumentException("Consumer ID cannot be null or empty");
        }
        
        if (messageId == null || messageId.isEmpty()) {
            throw new IllegalArgumentException("Message ID cannot be null or empty");
        }
        
        Topic topic = getTopic(topicName);
        return topic.acknowledge(consumerId, messageId);
    }
    
    @Override
    public int getMessageCount(String topicName) throws TopicNotFoundException {
        checkRunning();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        Topic topic = getTopic(topicName);
        return topic.getMessageCount();
    }
    
    @Override
    public void shutdown() {
        if (!running) {
            return;
        }
        
        running = false;
        
        topicsLock.writeLock().lock();
        try {
            for (Topic topic : topics.values()) {
                topic.close();
            }
            topics.clear();
            logger.info("MessageQueue shutdown successfully");
        } finally {
            topicsLock.writeLock().unlock();
        }
    }
    
    // 检查消息队列是否正在运行
    private void checkRunning() {
        if (!running) {
            throw new IllegalStateException("MessageQueue is not running");
        }
    }
    
    // 获取指定的主题，如果不存在则抛出异常
    private Topic getTopic(String topicName) throws TopicNotFoundException {
        topicsLock.readLock().lock();
        try {
            Topic topic = topics.get(topicName);
            if (topic == null) {
                throw new TopicNotFoundException(topicName);
            }
            return topic;
        } finally {
            topicsLock.readLock().unlock();
        }
    }
}